package ext.opensource.netty.client.mqtt.api.impl;

import java.util.ArrayList;
import java.util.List;

import ext.opensource.netty.client.core.BaseClient;
import ext.opensource.netty.client.mqtt.api.ClientProcess;
import ext.opensource.netty.client.mqtt.api.MqttConsumer;
import ext.opensource.netty.client.mqtt.api.MqttConsumerListener;
import ext.opensource.netty.client.mqtt.api.MqttConsumerProcess;
import ext.opensource.netty.client.mqtt.common.MessageData;
import ext.opensource.netty.client.mqtt.common.MessageStatus;
import ext.opensource.netty.client.mqtt.common.SubscribeMessage;
import ext.opensource.netty.client.mqtt.protocol.ClientProtocolUtil;
import ext.opensource.netty.common.MqttProtocolUtil;
import ext.opensource.netty.common.NettyLog;
import ext.opensource.netty.common.core.CacheList;
import ext.opensource.netty.common.core.CacheListLocalMemory;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.NonNull;
import lombok.Setter;

public class MqttConsumerProcessImpl extends ClientProcess implements MqttConsumerProcess, MqttConsumer {
	private MqttConsumerListener listener;
	@Setter
	@NonNull
	private CacheList<MessageData> msgListCache;

	public MqttConsumerProcessImpl(final BaseClient client) {
		this(client, new CacheListLocalMemory<MessageData>());
	}

	public MqttConsumerProcessImpl(final BaseClient client, final CacheList<MessageData> msgListCache) {
		super(client);
		this.msgListCache = msgListCache;
	}

	@Override
	public void subscribe(final String topic, final int qosValue) {
		final SubscribeMessage msgObj = SubscribeMessage.builder().topic(topic).qos(qosValue).build();
		subscribe(msgObj);
	}

	public void subscribe(final SubscribeMessage info) {
		NettyLog.debug("subscribe： {} ", info);
		subscribe(new SubscribeMessage[] { info });
	}

	private void subscribe(final SubscribeMessage... info) {
		if (info != null) {
			final int messageId = messageId().getNextMessageId(this.getClientId());
			if (!channel().isActive()) {
				NettyLog.debug("channel is close");
				return;
			}

			channel().writeAndFlush(ClientProtocolUtil.subscribeMessage(messageId, info));
		}
	}

	public void unSubscribe(final String topic) {
		final List<String> topics = new ArrayList<String>();
		topics.add(topic);
		unSubscribe(topics);
	}

	public void unSubscribe(final List<String> topics) {
		if (topics != null) {
			final int messageId = messageId().getNextMessageId(this.getClientId());
			channel().writeAndFlush(MqttProtocolUtil.unSubscribeMessage(topics, messageId));
		}
	}

	@Override
	public void saveMesage(final MessageData recviceMessage) {
		//if (msgListCache == null) {return;}
		if ((recviceMessage != null) && (recviceMessage.getMessageId() > 0)) {
			NettyLog.debug("saveMessage: {}", recviceMessage.getMessageId());
			msgListCache.put(String.valueOf(recviceMessage.getMessageId()), recviceMessage);
		}
	}

	@Override
	public void delMesage(final int messageId) {
		if (msgListCache == null) {
			return;
		}
		if (messageId > 0) {
			NettyLog.debug("delMesage: {}", messageId);
			msgListCache.remove(String.valueOf(messageId));
		}
	}

	public MessageData changeMessageStatus(final int messageId, final MessageStatus status) {
		if (msgListCache == null) {
			return null;
		}
		NettyLog.debug("changeMessageStatus: {}", status.name());

		final MessageData msgObj = msgListCache.get(String.valueOf(messageId));
		if (msgObj != null) {
			msgObj.setStatus(status);
			msgListCache.put(String.valueOf(messageId), msgObj);
		}
		return msgObj;
	}

	@Override
	public void processPubRel(final int messageId) {
		final MessageData msgObj = changeMessageStatus(messageId, MessageStatus.PUBREL);

		if ((msgObj != null) && (listener != null)) {
			if (msgObj.getQos() == MqttQoS.EXACTLY_ONCE.value()) {
				listener.receiveMessage(msgObj.getMessageId(), msgObj.getTopic(), decode(msgObj.getPayload()));
			}
		}
		NettyLog.debug("process Pub-rel: messageId - {} ", messageId);

	}

	@Override
	public void processSubAck(final int messageId) {
		NettyLog.debug("process Sub-ack: messageId - {} ", messageId);
	}

	@Override
	public void processUnSubBack(final int messageId) {
		NettyLog.debug("process Un-sub-back: messageId - {} ", messageId);
	}

	@Override
	public void processPublish(final MessageData msg) {
		NettyLog.debug("process Publish: {} ", msg);

		if (listener != null) {
			listener.receiveMessageByAny(msg.getMessageId(), msg.getTopic(), decode(msg.getPayload()));

			if (msg.getQos() == MqttQoS.AT_MOST_ONCE.value() || msg.getQos() == MqttQoS.EXACTLY_ONCE.value()) {
				listener.receiveMessage(msg.getMessageId(), msg.getTopic(), decode(msg.getPayload()));
			}
		}
	}

	@Override
	public void sendPubRecMessage(final int messageId) {
		NettyLog.debug("send Pub-rec: messageId - {} ", messageId);
		channel().writeAndFlush(MqttProtocolUtil.pubRecMessage(messageId));
	}

	@Override
	public void sendPubAckMessage(final int messageId) {
		NettyLog.debug("send Pub-ack: messageId - {} ", messageId);
		channel().writeAndFlush(MqttProtocolUtil.pubAckMessage(messageId));
	}

	@Override
	public void sendPubCompMessage(final int messageId) {
		NettyLog.debug("send Pub-comp: messageId - {} ", messageId);
		channel().writeAndFlush(MqttProtocolUtil.pubCompMessage(messageId));
	}

	@Override
	public void setConsumerListener(final MqttConsumerListener listener) {
		this.listener = listener;
	}

	@Override
	public void setCacheList(final CacheList<MessageData> msgList) {
		if (msgList != null) {
			this.msgListCache = msgList;
		}
	}
}
